package com.iot.mqtt.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.iot.common.utils.DateUtils;
import com.iot.common.utils.DictUtils;
import com.iot.common.utils.HexUtils;
import com.iot.common.utils.StringUtils;
import com.iot.common.utils.spring.SpringUtils;
import com.iot.device.domain.QuakeEvent;
import com.iot.device.domain.QuakeClient;
import com.iot.device.domain.QuakeEventOrigin;
import com.iot.device.service.IQuakeEventService;
import com.iot.device.service.IQuakeClientService;
import com.iot.device.service.IQuakeEventOriginService;
import com.iot.device.utils.LocationUtil;
import com.iot.mqtt.callback.QuakeClientMqttCallback;
import com.iot.websocket.Topic;
import com.iot.websocket.WebSocketMessageSender;
import com.iot.websocket.message.EventMsgMessage;
import com.iot.websocket.message.GpsMessage;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 设备消息处理器
 * @author Cao.ning
 * @date 2022/11/11
 */
@Component
public class QuakeClientHandler {
    private static final Logger logger = LoggerFactory.getLogger(QuakeClientHandler.class);

    @Autowired
    public ThreadPoolTaskExecutor threadPoolTaskExecutor;
    @Autowired
    public IQuakeClientService quakeClientService;
    @Autowired
    public IQuakeEventService quakeEventService;
    @Autowired
    public IQuakeEventOriginService quakeEventOriginService;

    /**
     * 处理消息
     * @param topic
     * @param message
     */
    public void handleMessage(String topic, MqttMessage message){
        // 必须用线程池处理，mqtt消息是单线程，处理完一个后才处理下一个，避免消息多了处理不了多余消息
        threadPoolTaskExecutor.execute(new Runnable() {
            @Override
            public void run() {
                logger.info("收到震动监测设备消息：{}， {}", topic, message);
                try {
                    if(message == null){
                        return;
                    }
                    String clientNo = topic.split("/GET")[0];
                    if(StringUtils.isEmpty(clientNo)){
                        return;
                    }
                    Date now = new Date();
                    QuakeClient quakeClient = quakeClientService.selectQuakeClientByClientNo(clientNo);
                    if(quakeClient == null){
                        // 如果是新注册的设备
                        quakeClient = new QuakeClient();
                        quakeClient.setClientNo(clientNo);
                        quakeClient.setAddTime(now);
                        quakeClient.setLastTime(now);
                        quakeClient.setOnlineTimes(0L);
                        // quakeClient.setRemark("设备自动注册");
                        quakeClient.setAddress("-");
                        quakeClient.setClientName(clientNo);
                        quakeClient.setCreateBy("register");
                        quakeClient.setUserId(1L);
                        quakeClient.setOrgId(200L);
                        quakeClient.setLongitude(new BigDecimal(120.421430));
                        quakeClient.setLatitude(new BigDecimal(36.275984));
                        quakeClientService.insertQuakeClient(quakeClient);
                    }
                    AlertEventData alertEventData = QuakeClientHandler.analyseData(quakeClient, new String(message.getPayload(), "UTF-8"));
                    if(alertEventData != null && alertEventData.getQuakeEventOrigin() != null){
                        quakeEventOriginService.insertQuakeEventOrigin(alertEventData.getQuakeEventOrigin());
                    }
                    if(alertEventData != null && !alertEventData.getEventList().isEmpty()){
                        int c = quakeEventService.batchInsertAlertEvent(alertEventData.getEventList());
                    }
                    alertEventData = null;
                    quakeClient.setLastTime(now);
                    quakeClientService.updateQuakeClient(quakeClient);
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
            }
        });
    }



    public static class AlertEventData {
        private QuakeEventOrigin eventOrigin;
        private QuakeEvent alertEvent;
        private List<QuakeEvent> eventList = new ArrayList<>();

        public AlertEventData() {
        }

        public AlertEventData(QuakeEventOrigin eventOrigin) {
            this.eventOrigin = eventOrigin;
        }

        public QuakeEventOrigin getQuakeEventOrigin() {
            return eventOrigin;
        }

        public void setQuakeEventOrigin(QuakeEventOrigin eventOrigin) {
            this.eventOrigin = eventOrigin;
        }

        public QuakeEvent getAlertEvent() {
            return alertEvent;
        }

        public void setAlertEvent(QuakeEvent alertEvent) {
            this.alertEvent = alertEvent;
        }

        public List<QuakeEvent> getEventList() {
            return eventList;
        }

        public void setEventList(List<QuakeEvent> eventList) {
            this.eventList = eventList;
        }

        public void addEvent(QuakeEvent event) {
            this.eventList.add(event);
        }
    }

    public static AlertEventData analyseData(QuakeClient quakeClient, String message){
        if(message.contains("cmdcnt=")){
            //
            return null;
        }
//        System.err.println(quakeClient.getClientNo() + "：" + message);

        // message消息格式
        /* angleZ=7.136200,angleX=0.062229,angleY=0.425780,event_ID=72905 */
        /* positionY=-37.699024,positionX=0.000000,positionZ=0.133742,event_ID=73058 */
        /* Ii=6.284557,event_ID=73056 */
        /*
            positionZ=22.766176,positionX=0.000000,positionY=-0.483218,event_ID=73051
            Ii=6.120072,event_ID=73053
         */
        Date now = DateUtils.getNowDate();
        quakeClient.setLastTime(now);
        if(HexUtils.isHex(message)){
            // 原始数据
            if(message.startsWith("5AA5")){
                message = message.substring(4, message.length()).replace(" ", "").replace("\r", "").replace("\n", "");
                byte[] bytes = HexUtils.hexStringToByte(message);
                JSONArray floats = HexUtils.bytes2FloatArray(bytes);
                QuakeEventOrigin eventOrigin = new QuakeEventOrigin();
                eventOrigin.setEventId(1L);//TODO 这里需要获取eventId 上传的数据等待 冉 修改
                eventOrigin.setClientNo(quakeClient.getClientNo());
                eventOrigin.setData(floats.toJSONString());
                eventOrigin.setHexData(message);
                return new AlertEventData(eventOrigin);
            }
        }
        // 指令回复
        if(message.contains("----------------------") || message.equals("restore PARA")
                || message.startsWith("admin,cclk") || message.startsWith("admin,coord")
                || message.startsWith("accu_") || message.startsWith("len=") || message.startsWith("shift_threshold="))
        {
            if(message.startsWith("admin,coord")){
                String[] msgs = message.split(",");
                if(msgs.length > 3 && StringUtils.isNotBlank(msgs[2]) && StringUtils.isNotBlank(msgs[3])){
                    String longitude = msgs[2];
                    String latitude = msgs[3];
                    if(StringUtils.isEmpty(quakeClient.getAddress()) || "-".equals(quakeClient.getAddress())){
                        com.alibaba.fastjson.JSONObject location = LocationUtil.baiduGeocoder(Double.parseDouble(longitude), Double.parseDouble(latitude)).getJSONObject("result");
                        StringBuilder address = new StringBuilder();
                        address.append(location.getString("formatted_address"))
                                .append("(")
                                .append(location.getString("sematic_description"))
                                .append(")");
                        quakeClient.setAddress(address.toString());
                    }
                    quakeClient.setLongitude(new BigDecimal(longitude));
                    quakeClient.setLatitude(new BigDecimal(latitude));
                    JSONObject json = new JSONObject();
                    json.put("longitude", longitude);
                    json.put("latitude", latitude);
                    WebSocketMessageSender.send(GpsMessage.getInstance(Topic.QUAKE_GPS, quakeClient.getClientNo(), json));
//                    SpringUtils.getBean(IQuakeClientService.class).updateQuakeClient(quakeClient);
                }
            }
            if((message.indexOf("cmdcnt")>=0)||(message.indexOf("paraflash")>=0)){
                /*ClientUpgradeTask.Result result = ClientUpgradeTask.STAT_MAP.get(quakeClient.getClientNo());
                if(result != null){
                    result.setReceived(true);
                    ClientUpgradeTask.STAT_MAP.put(quakeClient.getClientNo(), result);
                    logger.debug("收到更新响应，clientNo: {}, message: {}", quakeClient.getClientNo(), message);
                }*/
            }
//            WebSocketManager.s(quakeClient.getOrgId(), MessageTypeEnum.CLIENT_DEBUG_MESSAGE.getValue() + quakeClient.getClientNo(), message);
//            WebSocketManager.sendMessageCallback2Admin(MessageTypeEnum.CLIENT_DEBUG_MESSAGE.getValue() + quakeClient.getClientNo(), message);
            return null;
        }

        // 报警数据
        if(message.startsWith("position") || message.startsWith("angle") || message.startsWith("Ii")){
            String[] msgs = message.split("\\r\\n");
            boolean sendWsMsg = false;

            QuakeEvent.Position position = null;
            QuakeEvent.Angle angle = null;
            QuakeEvent.Ii ii = null;
//            QuakeClientBattery clientBattery = null;
            for(String msg : msgs){
                // msg格式：positionZ=22.766176,positionX=0.000000,positionY=-0.483218,event_ID=73051
                String[] record = msg.split(",");
                if(StringUtils.isBlank(msg) || record[0].equals("AAA_sum")){
                    continue;
                }

                String originEventId = null;
                for(String m : record){
                    String[] data = m.split("=");
                    String key = data[0];
                    if(data.length < 2){
                        continue;
                    }
                    String val = data[1];
                    if(key.equals("event_ID"))
                    {
                        originEventId = val;
                    }
                    // 位移
                    if(key.startsWith("position"))
                    {
                        if(position == null){
                            position = new QuakeEvent.Position(0f, 0f, 0f);
                            position.setOriginEventId(originEventId);
                        }
                        if(key.equals("positionX")){
                            position.setX(position.getX() + Float.parseFloat(val));
                        }else if(key.equals("positionY")){
                            position.setY(position.getY() + Float.parseFloat(val));
                        }else if(key.equals("positionZ")){
                            position.setZ(position.getZ() + Float.parseFloat(val));
                        }
                        continue;
                    }
                    // 倾斜
                    if(key.startsWith("angle"))
                    {
                        if(angle == null){
                            angle = new QuakeEvent.Angle(0f, 0f, 0f);
                            angle.setOriginEventId(originEventId);
                        }
                        if(key.equals("angleX")){
                            angle.setX(angle.getX() + Float.parseFloat(val));
                        }else if(key.equals("angleY")){
                            angle.setY(angle.getY() + Float.parseFloat(val));
                        }else if(key.equals("angleZ")){
                            angle.setZ(angle.getZ() + Float.parseFloat(val));
                        }
                        // 发生倾斜后重启，避免之后数据误报
                        SpringUtils.getBean(QuakeClientMqttCallback.class).publish(String.format(QuakeClientMqttCallback.TOPIC_SET, quakeClient.getClientNo()), 0, "*q\r\n");
                        continue;
                    }
                    // 震动
                    if(key.startsWith("Ii"))
                    {
                        if(ii == null){
                            ii = new QuakeEvent.Ii(0f);
                            ii.setOriginEventId(originEventId);
                        }
                        ii.setValue(ii.getValue() + Float.parseFloat(val));
                        continue;
                    }
                    // 电压
                    if(key.toLowerCase().equals("vbat"))
                    {
                        //Vbat=11.393702,Temp=31.125000
                        sendWsMsg = true;
//                        if(clientBattery == null){
//                            clientBattery = new QuakeClientBattery();
//                        }
//                        clientBattery.setClientNo(quakeClient.getClientNo());
//                        clientBattery.setVbat(Float.parseFloat(val));
//                        WebSocketMessageSender.send(GpsMessage.getInstance(Topic.GAS_GPS, gasClient.getClientNo(), json));
                    }
                    // 温度
                    if(key.toLowerCase().equals("temp"))
                    {
                        //Vbat=11.393702,Temp=31.125000
                        sendWsMsg = true;
//                        if(clientBattery == null){
//                            clientBattery = new QuakeClientBattery();
//                        }
//                        clientBattery.setClientNo(quakeClient.getClientNo());
//                        clientBattery.setTemp(Float.parseFloat(val));
                    }
                    if(key.equals("flag") || key.equals("raw_flag") || key.equals("port") || key.startsWith("tune_") || key.startsWith("shift_")){
                        sendWsMsg = true;
                    }
                }
                /*if(sendWsMsg){
                    WebSocketManager.sendMessageCallback(quakeClient.getOrgId(), MessageTypeEnum.CLIENT_DEBUG_MESSAGE.getValue() + quakeClient.getClientNo(), msg);
                    WebSocketManager.sendMessageCallback2Admin(MessageTypeEnum.CLIENT_DEBUG_MESSAGE.getValue() + quakeClient.getClientNo(), msg);
                }
                if(clientBattery != null){
                    // 保存电池电压温度信息记录
                    clientBattery.setClientNo(quakeClient.getClientNo());
                    clientBattery.setCreateTime(DateUtils.getNowDate());
                    SpringUtils.getBean(IQuakeClientBatteryService.class).insertQuakeClientBattery(clientBattery);
                    WebSocketManager.sendMessageCallback(quakeClient.getOrgId(), MessageTypeEnum.BATTERY_STATUS_MESSAGE.getValue(), clientBattery);
                    WebSocketManager.sendMessageCallback2Admin(MessageTypeEnum.BATTERY_STATUS_MESSAGE.getValue(), clientBattery);
                }*/
            }
            // 判断开启了的报警类型
            List<String> alertTypes = new ArrayList<>();
            if(DictUtils.getDictCache("alertType") != null){
                alertTypes = DictUtils.getDictCache("alertType").stream().map(item -> item.getDictValue()).collect(Collectors.toList());
            }
            if(!alertTypes.contains("Ii")){
                ii = null;
            }
            if(!alertTypes.contains("position")){
                position = null;
            }
            if(!alertTypes.contains("angle")){
                angle = null;
            }
            AlertEventData alertEventData = null;
            if(ii != null)
            {
                if(alertEventData == null){
                    alertEventData = new AlertEventData();
                }
                QuakeEvent quakeEvent = new QuakeEvent();
                quakeEvent.setEventTime(now);
                quakeEvent.setEventDate(now);
                quakeEvent.setClientNo(quakeClient.getClientNo());
                quakeEvent.setChannel(2L);
                quakeEvent.setDeptId(quakeClient.getOrgId());
                quakeEvent.setOriginData(message);
                quakeEvent.setAlertValue(ii.getValue());
                quakeEvent.setEventType("Ii");
                alertEventData.addEvent(quakeEvent);
            }
            if(position != null)
            {
                if(alertEventData == null){
                    alertEventData = new AlertEventData();
                }
                QuakeEvent quakeEvent = new QuakeEvent();
                quakeEvent.setEventTime(now);
                quakeEvent.setEventDate(now);
                quakeEvent.setClientNo(quakeClient.getClientNo());
                quakeEvent.setChannel(2L);
                quakeEvent.setDeptId(quakeClient.getOrgId());
                quakeEvent.setOriginData(message);
                quakeEvent.setEventType("position");
                quakeEvent.setAlertValue(Math.max(Math.max(Math.abs(position.getX()), Math.abs(position.getY())), Math.abs(position.getZ())));
                alertEventData.addEvent(quakeEvent);
            }
            if(angle != null)
            {
                if(alertEventData == null){
                    alertEventData = new AlertEventData();
                }
                QuakeEvent quakeEvent = new QuakeEvent();
                quakeEvent.setEventTime(now);
                quakeEvent.setEventDate(now);
                quakeEvent.setClientNo(quakeClient.getClientNo());
                quakeEvent.setChannel(2L);
                quakeEvent.setDeptId(quakeClient.getOrgId());
                quakeEvent.setOriginData(message);
                quakeEvent.setAlertValue(Math.max(Math.max(Math.abs(angle.getX()), Math.abs(angle.getY())), Math.abs(angle.getZ())));
                quakeEvent.setEventType("angle");
                alertEventData.addEvent(quakeEvent);
            }
            /*if(alertEventData != null && !alertEventData.getEventList().isEmpty()) {
                int c = iQuakeEventService.batchInsert(quakeEventList);
            }*/
            if(alertEventData != null)
            {
                QuakeEvent quakeEvent = new QuakeEvent();
                quakeEvent.setClientNo(quakeClient.getClientNo());
                quakeEvent.setEventTime(now);
                quakeEvent.setEventDate(now);
                quakeEvent.setClientNo(quakeClient.getClientNo());
                quakeEvent.setChannel(2L);
                quakeEvent.setDeptId(quakeClient.getOrgId());
                quakeEvent.setOriginData(message);
                if(ii != null){
                    quakeEvent.setAlertValue(ii.getValue());
                    quakeEvent.setEventType("Ii");
                    alertEventData.setAlertEvent(quakeEvent);
                }else if(position != null){
                    quakeEvent.setEventType("position");
                    quakeEvent.setAlertValue(Math.max(Math.max(Math.abs(position.getX()), Math.abs(position.getY())), Math.abs(position.getZ())));
                    alertEventData.setAlertEvent(quakeEvent);
                }else if(angle != null){
                    quakeEvent.setAlertValue(Math.max(Math.max(Math.abs(angle.getX()), Math.abs(angle.getY())), Math.abs(angle.getZ())));
                    quakeEvent.setEventType("angle");
                    alertEventData.setAlertEvent(quakeEvent);
                }
                quakeEvent.setClient(quakeClient);
                WebSocketMessageSender.send(EventMsgMessage.getInstance(Topic.QUAKE_EVENT_MSG, quakeEvent.getDeptId(), JSON.parseObject(JSON.toJSONString(quakeEvent))));
            }
            return alertEventData;
        }
        return null;
    }
}
